package operator;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.operators.shipping.OutputCollector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.api.operators.TimestampedCollector;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.util.Collector;

/**
 * FlatMap 算子
 */
public class FlatMapDemo1 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        // 1,2,3,4,5,6,7
        DataStreamSource<String> data = env.socketTextStream("192.168.56.10", 8888);

        SingleOutputStreamOperator<Integer> nums = data.transform("MyFlatMap", Types.INT, new MyStreamFlatMap<String, Integer>((String s, Collector<Integer> collector) -> {
            String[] words = s.split(",");
            for (String word : words) {
                int i = Integer.parseInt(word);
                if (i % 2 == 0) {
                    collector.collect(i);
                }
            }
        }));

        nums.print();

        env.execute("FlatMap Demo");
    }

    // 自定义StreamFlatMap
    public static class MyStreamFlatMap<String, Integer> extends AbstractUdfStreamOperator<Integer, FlatMapFunction<String, Integer>> implements OneInputStreamOperator<String, Integer> {

        public MyStreamFlatMap(FlatMapFunction<String, Integer> userFunction) {
            super(userFunction);
        }

        @Override
        public void processElement(StreamRecord<String> element) throws Exception {
            TimestampedCollector<Integer> collector = new TimestampedCollector<Integer>(output);
            userFunction.flatMap(element.getValue(), collector);
        }
    }
}
